package net.bwie.realtime.newcar.dwd.log.job;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.newcar.dwd.log.util.GeocodeUtil;
import net.bwie.realtime.newcar.utils.KafkaUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class DataCleanJob {
    public static void main(String[] args) throws Exception {
        // 1.env执行环境
        LocalStreamEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
        env.setParallelism(1);
        // 2.消费Kafka
        DataStream<String> kafkaData = KafkaUtil.consumerKafka(env, "vehicle_realtime");
        kafkaData.print();
        // 3.数据转换
        DataStream<String> processStream = processLog(kafkaData);
        processStream.print();
        // 4.输出到Kafka
        KafkaUtil.producerKafka(processStream,"new_car_clean_log");
        // 5.触发执行
        env.execute("DataCleanJob");
    }

    private static DataStream<String> processLog(DataStream<String> stream) {
        // 数据清洗
        DataStream<String> cleanStream = dataClean(stream);
        // 过滤异常数据
        DataStream<String> filterStream = dataFilter(cleanStream);
        // 经纬度转换省份
        DataStream<String> provinceStream = province(filterStream);
        // 过滤省份为未知和[]的数据
        DataStream<String> validProvinceStream = filterInvalidProvince(provinceStream);
        return validProvinceStream;
    }

    private static DataStream<String> filterInvalidProvince(DataStream<String> stream) {
        SingleOutputStreamOperator<String> province = stream.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                String province = jsonObject.getString("province");

                // 保留省份有效的数据（既不是"未知"也不是"[]"）
                return province != null && !"未知".equals(province) && !"[]".equals(province);
            }
        });
        return province;
    }

    private static DataStream<String> province(DataStream<String> stream) {
        SingleOutputStreamOperator<String> map = stream.map(new MapFunction<String, String>() {
            @Override
            public String map(String s) throws Exception {
                JSONObject json = JSON.parseObject(s);
                Double lon = json.getDouble("longitude");
                Double lat = json.getDouble("latitude");

                String province = "未知";
                if (lon != null && lat != null) {
                    province = GeocodeUtil.convertToProvince(lon, lat);
                }
                json.put("province", province);

                json.remove("longitude");
                json.remove("latitude");

                // 打印转换后的完整数据，确认是否有province字段
                String result = json.toJSONString();
                System.out.println("带省份的数据: " + result);
                return result;
            }
        });

        return map;
    }

    private static DataStream<String> dataFilter(DataStream<String> stream) {
        SingleOutputStreamOperator<String> filterData = stream.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                String vin = jsonObject.getString("vin");
                Long longitude = jsonObject.getLong("longitude");
                Long latitude = jsonObject.getLong("latitude");
                Long timestamp = jsonObject.getLong("timestamp");
                if (vin == null || vin.isEmpty() || timestamp == null || longitude == null || latitude == null) {
                    return false;
                }
                return true;
            }
        });
        return filterData;
    }

    private static DataStream<String> dataClean(DataStream<String> stream) {
        // 存放错误日志
        OutputTag<String> dirtyLog = new OutputTag<String>("dirty") {
        };
        SingleOutputStreamOperator<String> cleanData = stream.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
                try {
                    // 解析JSON
                    JSON.parseObject(value);
                    // 主流输出
                    out.collect(value);
                } catch (Exception e) {
                    // 存入测流
                    ctx.output(dirtyLog, value);
                }
            }
        });
        SideOutputDataStream<String> dirty = cleanData.getSideOutput(dirtyLog);
        KafkaUtil.producerKafka(dirty,"dirty_log");
        return cleanData;
    }
}